package com.gugee.tiktok.data.consumer.comment.listener;

import com.alibaba.nacos.api.config.annotation.NacosValue;
import com.gugee.tiktok.data.common.msg.CommentMessage;
import com.gugee.tiktok.data.consumer.comment.config.CommentAppConfig;
import com.gugee.tiktok.data.consumer.comment.core.main.SaveAwemeCommentMain;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;

/**
 *
 * @author luolifeng
 * Date  2020/10/27
 */
@Component
@Slf4j
public class CommentMessageListener {
    @Autowired
    SaveAwemeCommentMain saveAwemeCommentMain;

    @Autowired
    CommentAppConfig commentAppConfig;
    @KafkaListener(id="${kafka.consumer.group.comment.id}",topics = {"${kafka.topics.comment.online}"},containerFactory = "factoryCommentAck")
    public void listen(List<CommentMessage> messages, Acknowledgment ack) {
        long start=System.currentTimeMillis();
        long acceptMsgTimeSecRange=commentAppConfig.getAcceptMsgTimeMinRange()*60*1000;
        String error="success";
        try {

            long isRange=start-acceptMsgTimeSecRange;
            final int[] noInRangeCount = {0};
            List<CommentMessage>  cmss= messages.stream().filter(msg->{
                /**
                 * 消息如果不在处理时限内，则丢弃
                 */
                if(msg.getEventTime() < isRange) {
                    noInRangeCount[0]++;
                    return false;
                }
                //视频已有评论数小于最低要求则先不爬取
                if (msg.getOCommentCount() < commentAppConfig.getCollectMinComments()) {
                    return false;
                }
                //s3中存的评论数

                int s3Comments = msg.getSynComments();
                //视频已收录评论数大于1000则不再爬取
                if (s3Comments >= commentAppConfig.getCollectMaxComments()) {
                    return false;
                }

                //评论数已采满
                if (s3Comments==msg.getOCommentCount()) {
                    return false;
                }
                if(msg.getId()==null || msg.getId()==0){
                    log.error("接收到的消息无id值，OAwemeId={}",msg.getOAwemeId());
                    return false;
                }

                return true;
            }).collect(Collectors.toList());
            if(noInRangeCount[0] > 0){
                log.info("接收到评论消息数量: {},需解析评论消息数量： {},不在处理时限内消息数量： {},时限分钟：{}" , messages.size(),cmss.size(),noInRangeCount[0],commentAppConfig.getAcceptMsgTimeMinRange());

            }
            saveAwemeCommentMain.handle(cmss);



            ack.acknowledge();
        }catch (Exception e){
            log.error("发生异常：{}" ,e.getMessage());
            error=e.getMessage();
        }finally {

            long times=(System.currentTimeMillis()-start)/1000;
            if(times<1){
                times=1;
            }
            log.info("评论消息数量: {},处理耗时: {}s,每秒处理消息数={}/s,结果：{}" , messages.size(),times,messages.size()/times,error);
        }
    }

}
